{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "b3b90931",
   "metadata": {},
   "source": [
    "# 多动作单智能体\n",
    "\n",
    "## **具有多个动作的智能体**\n",
    "\n",
    "我们注意到一个智能体能够执行一个动作，但如果只有这些，实际上我们并不需要一个智能体。通过直接运行动作本身，我们可以得到相同的结果。智能体的力量，或者说Role抽象的惊人之处，在于动作的组合（以及其他组件，比如记忆，但我们将把它们留到后面的部分）。通过连接动作，我们可以构建一个工作流程，使智能体能够完成更复杂的任务。\n",
    "\n",
    "假设现在我们不仅希望用自然语言编写代码，而且还希望生成的代码立即执行。一个拥有多个动作的智能体可以满足我们的需求。让我们称之为RunnableCoder，一个既写代码又立即运行的Role。我们需要两个Action：SimpleWriteCode 和 SimpleRunCode。\n",
    "\n",
    "## **定义动作**\n",
    "\n",
    "首先，定义 SimpleWriteCode。我们将重用上面创建的那个。\n",
    "\n",
    "接下来，定义 SimpleRunCode。如前所述，从概念上讲，一个动作可以利用LLM，也可以在没有LLM的情况下运行。在SimpleRunCode的情况下，LLM不涉及其中。我们只需启动一个子进程来运行代码并获取结果。我们希望展示的是，对于动作逻辑的结构，我们没有设定任何限制，用户可以根据需要完全灵活地设计逻辑。\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "3bb51598",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-02-26 14:31:59.312 | INFO     | metagpt.const:get_metagpt_package_root:29 - Package root set to /tmp/code/wow-agent/notebooks\n"
     ]
    }
   ],
   "source": [
    "# SimpleWriteCode 这个类与上一节一模一样\n",
    "\n",
    "from metagpt.actions import Action\n",
    "\n",
    "class SimpleWriteCode(Action):\n",
    "    PROMPT_TEMPLATE: str = \"\"\"\n",
    "    Write a python function that can {instruction} and provide two runnnable test cases.\n",
    "    Return ```python your_code_here ```with NO other texts,\n",
    "    your code:\n",
    "    \"\"\"\n",
    "\n",
    "    name: str = \"SimpleWriteCode\"\n",
    "\n",
    "    async def run(self, instruction: str):\n",
    "        prompt = self.PROMPT_TEMPLATE.format(instruction=instruction)\n",
    "\n",
    "        rsp = await self._aask(prompt)\n",
    "\n",
    "        code_text = SimpleWriteCode.parse_code(rsp)\n",
    "\n",
    "        return code_text\n",
    "\n",
    "    @staticmethod\n",
    "    def parse_code(rsp):\n",
    "        pattern = r\"```python(.*)```\"\n",
    "        match = re.search(pattern, rsp, re.DOTALL)\n",
    "        code_text = match.group(1) if match else rsp\n",
    "        return code_text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "f001112c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 本节新增了SimpleRunCode这个类\n",
    "class SimpleRunCode(Action):\n",
    "    name: str = \"SimpleRunCode\"\n",
    "\n",
    "    async def run(self, code_text: str):\n",
    "        result = subprocess.run([\"python\", \"-c\", code_text], capture_output=True, text=True)\n",
    "        code_result = result.stdout\n",
    "        logger.info(f\"{code_result=}\")\n",
    "        return code_result"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d2608db4",
   "metadata": {},
   "source": [
    "## **定义角色**\n",
    "\n",
    "与定义单一动作的智能体没有太大不同！让我们来映射一下：\n",
    "\n",
    "1. 用 self.set\\_actions 初始化所有 Action\n",
    "2. 指定每次 Role 会选择哪个 Action。我们将 react\\_mode 设置为 \"by\\_order\"，这意味着 Role 将按照 self.set\\_actions 中指定的顺序执行其能够执行的 Action。在这种情况下，当 Role 执行 \\_act 时，self.rc.todo 将首先是 SimpleWriteCode，然后是 SimpleRunCode。\n",
    "3. 覆盖 \\_act 函数。Role 从上一轮的人类输入或动作输出中检索消息，用适当的 Message 内容提供当前的 Action (self.rc.todo)，最后返回由当前 Action 输出组成的 Message。\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "68e14df0",
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "import os\n",
    "import subprocess\n",
    "from metagpt.roles import Role\n",
    "from metagpt.schema import Message\n",
    "from metagpt.logs import logger\n",
    "class RunnableCoder(Role):\n",
    "    name: str = \"Alice\"\n",
    "    profile: str = \"RunnableCoder\"\n",
    "\n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        self.set_actions([SimpleWriteCode, SimpleRunCode])\n",
    "        self._set_react_mode(react_mode=\"by_order\")\n",
    "\n",
    "    async def _act(self) -> Message:\n",
    "        logger.info(f\"{self._setting}: to do {self.rc.todo}({self.rc.todo.name})\")\n",
    "        # By choosing the Action by order under the hood\n",
    "        # todo will be first SimpleWriteCode() then SimpleRunCode()\n",
    "        todo = self.rc.todo\n",
    "\n",
    "        msg = self.get_memories(k=1)[0]  # find the most k recent messages\n",
    "        result = await todo.run(msg.content)\n",
    "\n",
    "        msg = Message(content=result, role=self.profile, cause_by=type(todo))\n",
    "        self.rc.memory.add(msg)\n",
    "        return msg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "4ce001d9",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-02-26 14:32:41.080 | INFO     | __main__:main:4 - write a function that calculates the sum of a list\n",
      "2025-02-26 14:32:41.083 | INFO     | __main__:_act:17 - Alice(RunnableCoder): to do SimpleWriteCode(SimpleWriteCode)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "```python\n",
      "def sum_list(numbers):\n",
      "    return sum(numbers)\n",
      "\n",
      "# Test cases\n",
      "def test_case_1():\n",
      "    assert sum_list([1, 2, 3, 4, 5]) == 15\n",
      "\n",
      "def test_case_2():\n",
      "    assert sum_list([]) == 0\n",
      "```"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2025-02-26 14:32:44.236 | WARNING  | metagpt.utils.cost_manager:update_cost:49 - Model glm-4-flash not found in TOKEN_COSTS.\n",
      "2025-02-26 14:32:44.241 | INFO     | __main__:_act:17 - Alice(RunnableCoder): to do SimpleRunCode(SimpleRunCode)\n",
      "2025-02-26 14:32:44.276 | INFO     | __main__:run:8 - code_result=''\n",
      "2025-02-26 14:32:44.279 | INFO     | __main__:main:6 - RunnableCoder: \n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "RunnableCoder: \n"
     ]
    }
   ],
   "source": [
    "async def main():\n",
    "    msg = \"write a function that calculates the sum of a list\"\n",
    "    role = RunnableCoder()\n",
    "    logger.info(msg)\n",
    "    result = await role.run(msg)\n",
    "    logger.info(result)\n",
    "    return result\n",
    "\n",
    "rtn = await main()\n",
    "print(rtn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5735c67f",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
